package bx.job.actors;

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import bx.job.JobServerBootstrap;
import bx.job.lock.bean.ScheduleLockContextBean;
import bx.job.lock.bean.ScheduleUnlockContextBean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;

public class LockManagerActor extends AbstractBehavior<LockManagerActor.LockManagerActorMsg> {
    private final static Logger log = LoggerFactory.getLogger(JobScheduleActor.class);
    public static final int LOCKING = 1;
    public static final int LOCKED = 2;

    public static final int UNLOCKING = 3;
    public static final int UNLOCKED = 4;
    private int lockState;

    public LockManagerActor(ActorContext<LockManagerActorMsg> context) {
        super(context);
    }

    public static Behavior<LockManagerActor.LockManagerActorMsg> create() {
        return Behaviors.setup(LockManagerActor::new);
    }

    @Override
    public Receive<LockManagerActorMsg> createReceive() {
        return newReceiveBuilder().onMessage(LockManagerActor.LockMsg.class, lockMsg -> {
            if (lockState == LOCKING) {
                log.info(" The lock is processing, lock status is: {} ", lockState);
                return this;
            }

            if (lockState != UNLOCKED) {
                log.info("The current process is not complete, ignore this schedule. The lock status is:{} ", lockState);
                return this;
            }

            lockState = LOCKING;
            CompletableFuture.supplyAsync(() -> JobServerBootstrap.getScheduleLock().lockScheduler(ScheduleLockContextBean.builder().build())).thenAccept(lock -> {
                if (!lock) { // 不能获取锁，直接返回
                    log.error("The current lock process is not complete");
                    lockState = UNLOCKED;
                } else {
                    log.info("lock success");
                    lockState = LOCKED;
                    lockMsg.replyTo.tell(new JobScheduleActor.LockSuccessMsg(lockMsg.lockContext()));
                }
            });
            return this;
        }).onMessage(LockManagerActor.UnlockMsg.class, unlockMsg -> {
            if (lockState == UNLOCKING) {
                log.info(" The unlock is processing, lock status is: {} ", lockState);
                return this;
            }


            if (lockState != LOCKED) { // 理论上不应该出现这种状态，但有必要检测一下
                log.error(" The lock status is invalid :{} ", lockState);
                lockState = UNLOCKED;
                return this;
            }

            CompletableFuture.supplyAsync(() -> JobServerBootstrap.getScheduleLock().unlockScheduler(unlockMsg.unLockContext())).thenAccept(lock -> {
                if (!lock) {
                    log.error("unlock process is failed");
//                    lockState = UNLOCKED; // todo 因为网络问题 释放锁失败是否应该开启下一次流程？
                } else {
                    log.info("unlock success");
                    lockState = UNLOCKED;
                    unlockMsg.replyTo.tell(new JobProcessManagerActor.UnlockedMsg());
                }
            });
            return this;
        }).build();
    }

    public interface LockManagerActorMsg {
    }

    public static record LockMsg(ScheduleLockContextBean lockContext,
                                 ActorRef<JobScheduleActor.JobScheduleActorMsg> replyTo) implements LockManagerActorMsg {
    }

    public static record UnlockMsg(ScheduleUnlockContextBean unLockContext,
                                   ActorRef<JobProcessManagerActor.JobProcessManagerActorMsg> replyTo) implements LockManagerActorMsg {
    }

}
